package com.ksyun.campus.dataserver.services;

import com.ksyun.campus.dataserver.config.dataserverconfig.DataServerConfigProperties;
import com.ksyun.campus.dataserver.config.zookperconfig.ZookeeperConfigProperties;
import com.ksyun.campus.dataserver.domain.DataServerInfo;
import com.ksyun.campus.dataserver.domain.ReplicaData;
import com.ksyun.campus.dataserver.param.FileWriteParam;
import com.ksyun.campus.dataserver.util.FileUtil;
import com.ksyun.campus.dataserver.util.LoadBalanceUtil;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.*;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Recover;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestTemplate;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;

@SuppressWarnings("unchecked")
@Service
public class DataService {
    /**
     * 最小服务台数
     */
    private static final int SERVER_NUM = 3;
    @Autowired
    private ZkClient zkClient;
    @Autowired
    private DataServerConfigProperties dataServerConfigProperties;
    @Autowired
    private ZookeeperConfigProperties zookeeperConfigProperties;
    @Autowired
    private RestTemplate restTemplate;
    @Autowired
    private DataServerInfo dataServerInfo;

    /**
     * 失败重试次数
     */
    private static final int RETRY_TIMES = 3;
    @Autowired
    private Map<String, DataServerInfo> dataServerMap;

    /**
     * 写文件节点，并返回副本信息列表
     *
     * @param originPath 文件自己的全目录
     * @param data       数据
     * @param flag       是否为写从数据操作，为true时是写主数据，为false为写从数据
     * @return
     */
    public List<ReplicaData> write(String originPath, byte[] data, boolean flag) {
        List<String> dslist = zkClient.getChildren(dataServerConfigProperties.getFullNode());
        if (dslist.size() < SERVER_NUM) { // 不足三台机器时不写
            return null;
        }

        // 写入本地文件
        // 获取文件id 唯一
        String fileName = FileUtil.getIdFromByteArray(data);
        // 获取文件id 不唯一
//        String fileName = UUID.randomUUID().toString().replace("-", "");
//        FileUtil.writeByteArrayToFile(data, dataServerConfigProperties.getDataPath() + File.separator + originPath + File.separator + fileName);
        try {
            FileUtil.writeByteArrayToFile(data, dataServerConfigProperties.getDataPath() + File.separator + fileName);
        } catch (IOException e) {
            return null;
        }
        dataServerInfo.setUsedCapacity(dataServerInfo.getUsedCapacity() + data.length);
        // 调用远程ds服务写接口，同步副本，已达到多副本数量要求
        // 强一致写
        if (!flag) {
            return wrapLocalReplicaData(originPath, fileName);
        }

        List<ReplicaData> replicaDataList = wrapLocalReplicaData(originPath, fileName);
        // 选择策略，按照 az rack->zone 的方式选取，将三副本均分到不同的az下
        // 先按机架选，再按区域，做到读取时距离服务器较近
        LoadBalanceUtil loadBalance = new LoadBalanceUtil((List<DataServerInfo>) dataServerMap.values());
        List<DataServerInfo> loadBalanceServers = loadBalance.getLoadBalanceServersByCondition(SERVER_NUM - 1, dataServerInfo.getRack(), dataServerInfo.getZone());

        // 转发请求写副本
        for (DataServerInfo server : loadBalanceServers) {
            String targetUrl = "http://" + server.getIp() + ":" + server.getPort() + "/write";
            List<ReplicaData> rep = forwardWrite(targetUrl, originPath, data, false);
            if (rep != null && rep.size() != 0) {
                replicaDataList.add(rep.get(0));
            }
        }
        // 如果写失败，不足3副本，则退回写操作，删除本地
        if (replicaDataList.size() == SERVER_NUM) {
            // 返回三副本位置
            return replicaDataList;
        }
        FileUtil.deleteFile(dataServerConfigProperties.getDataPath() + File.separator + fileName);
        dataServerInfo.setUsedCapacity(dataServerInfo.getUsedCapacity() - data.length);
        return Collections.emptyList();
    }

    private List<ReplicaData> wrapLocalReplicaData(String originPath, String fileName) {
        List<ReplicaData> result = new CopyOnWriteArrayList<>();
        result.add(new ReplicaData(fileName, dataServerInfo.getIp() + ":" + dataServerInfo.getPort(), originPath));
        return result;
    }

    @Retryable(
            value = {Exception.class},
            // include = {},// 处理异常
            // exclude = {},// 例外异常
            maxAttempts = RETRY_TIMES, // 最大重试次数
            backoff = @Backoff(delay = 3000) // 重试间隔时间为 3 秒
    )
    public List<ReplicaData> forwardWrite(String targetUrl, String originPath, byte[] data, boolean flag) {
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.MULTIPART_FORM_DATA);

        MultiValueMap<String, Object> body = new LinkedMultiValueMap<>();
        FileWriteParam fileWriteParam = new FileWriteParam();
//        fileWriteParam.setFileSystem("yourFileSystem");
        fileWriteParam.setPath(originPath);
        fileWriteParam.setMainDataFlag(flag);

        body.add("fileWriteParam", fileWriteParam);
        body.add("data", new HttpEntity<>(data, headers)); // Add the file data

        HttpEntity<MultiValueMap<String, Object>> requestEntity = new HttpEntity<>(body, headers);

        ResponseEntity<List<ReplicaData>> responseEntity = restTemplate.exchange(
                targetUrl,
                HttpMethod.POST,
                requestEntity,
                new ParameterizedTypeReference<List<ReplicaData>>() {
                }
        );

        if (responseEntity.getStatusCode() == HttpStatus.CREATED) {
            return responseEntity.getBody();
        } else {
            throw new RuntimeException("Write operation failed with status: " + responseEntity.getStatusCode());
        }
    }


    /**
     * 重试失败后的兜底方法
     *
     * @param
     * @return
     */
    @Recover
    public List<ReplicaData> baseRetryRecover() {
        return Collections.emptyList();
    }

    public byte[] read(String id) {
        String path = dataServerConfigProperties.getDataPath() + File.separator + id;
        // todo 根据path读取指定大小的内容
        try {
            // 读取整个文件为字节数组
            byte[] bytes = Files.readAllBytes(Paths.get(path));
            return bytes;
        } catch (IOException e) {
            return null;
        }
    }


}
